from pyspark.sql import SparkSession
from pyspark.sql.functions import col

from SparkSessionBase import SparkSessionBase  # 假设这是你自己的封装类


# 继承 SparkSessionBase 类
class UserReviewApi(SparkSessionBase):
    SPARK_URL = "yarn"
    SPARK_APP_NAME = 'UserReviewApi'
    ENABLE_HIVE_SUPPORT = True

    def __init__(self):
        self.spark = self._create_spark_session()
        self.spark.sparkContext.setLogLevel("ERROR")  # 设置日志级别为 ERROR，减少日志

    def get_user_reviews(self, user_id):
        # 从 review 表中获取用户的所有评论
        review_df = self.spark.table('review')
        user_reviews = review_df.filter(col("rev_user_id") == user_id).select("*")

        return user_reviews

    def get_business_name(self, business_id):
        # 从 business 表中获取商家名称
        business_df = self.spark.table('business')
        business_name = business_df.filter(col("business_id") == business_id).select("name")
        return business_name

    def run(self, user_id):
        user_reviews = self.get_user_reviews(user_id)
        if not user_reviews.rdd.isEmpty():
            # 获取 business 表
            business_df = self.spark.table('business')
            # 将 user_reviews 和 business_df 按 rev_business_id 和 business_id 关联，增加 business_name 字段
            user_reviews_with_business_name = user_reviews.join(business_df, user_reviews.rev_business_id == business_df.business_id, "left").select(user_reviews["*"], business_df["name"].alias("business_name"))
            print("user reviews success")
            return user_reviews_with_business_name
        else:
            print(f"No reviews found for user with ID: {user_id}")
            return None


if __name__ == '__main__':
    # 示例调用
    job = UserReviewApi()
    result = job.run(user_id="9jh7Q3keoOkdASlpECJ-ig")
    if result is not None:
        result.show(truncate=False)